package com.wangyh8216.demo.flink;

// Flink类型系统相关导入
import org.apache.flink.api.common.typeinfo.Types;
// Flink元组类型
import org.apache.flink.api.java.tuple.Tuple2;
// Flink流处理数据流相关导入
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
// Flink流处理执行环境
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
// Flink数据收集器
import org.apache.flink.util.Collector;

/**
 * 流处理单词计数示例程序
 * 演示Flink流处理API的基本使用方法
 * 功能：实时统计文本文件中每个单词的出现次数
 * 
 * 与批处理版本(BatchWordCount)的主要区别：
 * 1. 使用StreamExecutionEnvironment而不是ExecutionEnvironment
 * 2. 数据流是持续处理的，需要显式调用env.execute()启动作业
 * 3. 支持实时数据流处理
 * 
 * @author wangyh8216
 * @version 1.0
 * @since 2024
 */
public class WordCount {

    /**
     * 主方法 - 流处理单词计数程序的入口点
     * 执行流程：
     * 1. 创建流处理执行环境
     * 2. 读取文本文件数据源
     * 3. 分词并转换为(单词, 1)元组
     * 4. 按单词进行键控分区
     * 5. 对每个单词的计数进行累加
     * 6. 输出实时统计结果
     * 7. 启动流处理作业
     * 
     * @param args 命令行参数（未使用）
     */
    public static void main(String[] args) {
        // 1、创建流处理执行环境
        // StreamExecutionEnvironment是Flink流处理作业的入口点
        // 与批处理不同，流处理环境支持实时数据流处理
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2、从文件读取数据源（虽然是文件，但按流处理方式读取）
        // DataStreamSource<String>表示字符串类型的流数据源
        // readTextFile方法会将文件内容作为数据流读取
        DataStreamSource<String> stringDataStreamSource = env.readTextFile("src/main/resources/words.txt");

        // 3、数据转换：将每行文本分词并转换为(单词, 1)二元组
        // flatMap操作将每行文本拆分为多个单词，并为每个单词生成计数1
        // SingleOutputStreamOperator表示单输出流的操作结果
        SingleOutputStreamOperator<Tuple2<String, Long>> wordTuple = stringDataStreamSource.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            // 使用空格分隔符将一行文本拆分为单词数组
            String[] words = line.split(" ");
            // 遍历每个单词，输出(单词, 1)元组
            for (String word : words) {
                // 收集器输出元组，格式为(单词, 计数1)
                out.collect(Tuple2.of(word, 1L));
            }
        // 显式指定返回类型，避免Java泛型类型擦除问题
        // Types.TUPLE定义元组类型，Types.STRING表示字符串类型，Types.LONG表示长整型
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 4、键控分区：按照单词字段进行分组
        // keyBy操作将数据流按指定键进行分区，相同键的数据会被发送到同一个任务槽
        // data -> data.f0 表示按元组的第一个字段（即单词）进行分区
        // KeyedStream表示键控分区后的数据流
        KeyedStream<Tuple2<String, Long>, String> keyedStream = wordTuple.keyBy(data -> data.f0);

        // 5、聚合统计：对每个键（单词）的计数字段进行累加
        // sum(1)表示对元组的第二个字段（即计数）进行累加操作
        // 流处理中的sum操作是状态化的，会持续累加相同键的值
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = keyedStream.sum(1);

        // 6、输出结果到控制台
        // print()方法会将数据流的内容打印到控制台
        // 在流处理中，print()不会立即触发执行，需要调用env.execute()
        sum.print();

        try {
            // 7、启动流处理作业执行
            // execute()方法会启动Flink作业，开始处理数据流
            // 与批处理不同，流处理作业会持续运行直到被显式停止
            env.execute();
        } catch (Exception e) {
            // 异常处理：打印异常堆栈信息
            e.printStackTrace();
        }
    }
}
